package com.xian.java.stream;

import com.xian.bean.Temperature;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 尚未测试完毕
 *
 */
public class ByKeyTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);
        //DataStreamSource<String> stringDataStreamSource = env.readTextFile("E:/test/Temperature.txt");
        DataStream<String> dataStream = env.readTextFile("E:/test/Temperature.txt"); //用父类接
        //dataStream.print();


        SingleOutputStreamOperator<Temperature> singleOutputStreamOperator = dataStream.map(new RichMapFunction<String, Temperature>() {
            @Override
            public Temperature map(String line) throws Exception {
                String[] arr = line.split(",");
                int id = Integer.parseInt(arr[0]);
                double t = Double.parseDouble(arr[1]);
                long ts = Long.parseLong(arr[2]);
                Temperature temperature = new Temperature(id, t, ts);
                return temperature;
            }
        });


        KeyedStream<Temperature, Tuple> temper = singleOutputStreamOperator.keyBy("temper");

        temper.print();


        env.execute();


    }
}
